feat: full native make_interval with 2.6x faster than spark and 1.1x faster than codegen dispatch - #5292
feat: full native make_interval with 2.6x faster than spark and 1.1x faster than codegen dispatch#5292peterxcli wants to merge 27 commits into
Conversation
peterxcli
changed the title
peterxcli
changed the title
andygrove
left a comment
andygrove
left a comment
There was a problem hiding this comment.
Thanks for the significant cleanup here — moving make_interval fully native and fixing the microsecond-precision loss is a real quality improvement, and the benchmark numbers show it. A few things worth addressing before merge.
ANSI overflow message diverges from Spark
Spark's MakeInterval.nullSafeEval catches the ArithmeticException from IntervalUtils.makeInterval and rethrows via arithmeticOverflowError(e.getMessage), where e.getMessage is Java's Math.addExact / multiplyExact string — literally "integer overflow" or "long overflow". Under ANSI, Spark reports:
[ARITHMETIC_OVERFLOW] integer overflow. If necessary set "spark.sql.ansi.enabled" to "false" to bypass this error.
Comet's kernel always calls arithmetic_overflow_error(""interval"") at native/spark-expr/src/datetime_funcs/make_interval.rs:150, producing interval overflow instead. I confirmed this with a targeted expect_error(integer overflow) on make_interval(2147483647) — Spark's message matches, Comet's doesn't. The fixtures in this PR use expect_error(overflow. If necessary set), which matches both and hides the divergence.
Could the free make_interval function return the specific label (""integer"" for the year/month and week/day paths, ""long"" for the microsecond path) so arithmetic_overflow_error gets the same string Spark produces? A tighter expect_error in make_interval_ansi.sql would then pin the parity going forward.
Utils.toArrowType CalendarIntervalType branch is now unreachable
With the new case CalendarIntervalType => branch in toArrowField at spark/src/main/scala/org/apache/spark/sql/comet/util/Utils.scala:212, the corresponding arm of toArrowType at line 172 is unreachable. If a future caller ever hits it directly, they get an unnamed, untagged ArrowType.Struct.INSTANCE with no children, and isCalendarIntervalStructField would fail to identify it on the round-trip. Would it be safer to make that arm throw with a message pointing callers at toArrowField, since the tagged struct can't be produced from an ArrowType alone?
The _dispatch SQL fixtures no longer exercise a distinct path
With MakeInterval unconditionally native, make_interval_dispatch.sql and make_interval_dispatch_ansi.sql run through the same kernel as make_interval.sql / make_interval_ansi.sql. The dispatch name is now misleading and the queries mostly duplicate coverage. Could these either be removed, or repurposed with -- Config: spark.comet.exec.enabled=false at the top so they cover the pure-Spark fallback path instead?
Argument downcasts .unwrap() where sibling kernels return errors
At native/spark-expr/src/datetime_funcs/make_interval.rs:109-118, the six Int32Array and one Decimal128Array downcasts unconditionally .unwrap(). This is unreachable today because the planner inserts a CastExpr for any input whose type differs from the Signature::exact, but SparkMakeDate in the same directory uses .ok_or_else(|| DataFusionError::Execution(...))? for the same pattern. Matching that style would give a diagnostic instead of a panic if a future serde change ever bypasses the cast.
Boundary test coverage: only positive extreme
The Int.MaxValue boundary row in make_interval.sql covers the positive side of the checked arithmetic. It might be worth adding a mirror row that pushes at least one of year/month/week/day toward Int.MinValue with a negative secs at the boundary, so the negative side of checked_add / checked_mul is validated against Spark too. Spark's own IntervalExpressionsSuite only tests the positive case, but with the native kernel the two sides are separate branches of i32::checked_mul and it's cheap to cover both.
End-to-end test for chained native consumption
calendar_interval.sql covers the shuffle case for array<interval>, but there's no test where a native make_interval result flows directly into another native step. If FFI (or any other stage on the interval column's path) ever drops the SPARK::calendarInterval::struct metadata on the months child, isCalendarIntervalStructField would silently degrade the type to an anonymous StructType(months, days, microseconds) and downstream CalendarInterval consumers would break with no clear error. A query like SELECT date '2020-01-01' + make_interval(1, 2, 3, 4, 5, 6, 7.123456) FROM t or a similar chained native consumer would pin the metadata-preservation invariant.
I ran CometSqlFileTestSuite make_interval locally on this branch — all 6 files pass. I also probed several extreme non-ANSI boundary cases (both signs of i32::MAX years/hours/minutes with -999999999999.999999 and +999999999999.999999 secs) and results match Spark. The change looks solid; the items above are all in-band for this PR.
|
@andygrove Thanks for the detailed review. I addressed each point:
Changed the helper to return I kept the SQL assertion as
Changed
Removed both dispatch fixtures. With
Replaced all seven argument-downcast
Added a row reaching
Added an end-to-end query that constructs |
|
@andygrove I've addressed your review, would appreciate it if you could take a look at the update and see if we can get this merged. |
Fixing the 1000x range loss and the Four things. Should the fix go upstream to This replaces the
Turning a working case into an Cross-version compatibility of the wire format The Arrow representation of Scope 23 files touching serde, FFI, Arrow readers and writers, codegen input and output, Scala UDF codegen, and the native kernel. That is a lot of surface for one review pass, and the two |
…Field Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
|
@andygrove Thanks for the second pass — replies inline, plus one doc improvement pushed. Upstream vs. fork: The tagged-struct representation is Comet-specific, so this can't be an upstream fix. In
Cross-version wire format: Not a supported scenario, because it can't arise: the Comet jar is fixed per Spark application, so all executors in one app run the same Comet build, and shuffle blocks (external shuffle service included) are application-scoped — nothing written by one app's Comet is ever read by a different version. Comet persists no Arrow-format state beyond shuffle lifetime; durable storage is Parquet. I've added a note to the description rather than the migration guide. Dispatch path: It's intentionally gone for Doc comment: Good call on |
…ke-interval # Conflicts: # native/spark-expr/Cargo.toml # spark/src/test/scala/org/apache/spark/sql/comet/execution/arrow/CometArrowStreamSuite.scala
|
@sunchao would you like to take a look at this and see if this can be merged? thanks! |
andygrove
added
enhancement
This branch moved `datafusion-spark` to `[dev-dependencies]` once the native `make_interval` stopped using `SparkMakeInterval`, keeping it only for the timestamp benches. Since then `main` added library uses of the crate in `string_funcs/concat_ws.rs` (apache#5725) and `agg_funcs/collect.rs`, so the merged crate failed to compile with `E0433: cannot find module or crate datafusion_spark`, breaking every native build, rust-test, the Spark SQL and Iceberg builds, the benchmark check and the Delta build gate. Move it back under `[dependencies]`, where `main` keeps it; benches still see it, so the dev-dependency entry is dropped. `Cargo.lock` is unchanged because it already recorded the dependency. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
…ke-interval # Conflicts: # native/spark-expr/src/lib.rs # spark/src/test/scala/org/apache/comet/CometCodegenSuite.scala
# Conflicts: # native/spark-expr/src/lib.rs
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Summary
- Prior state and problem: Calendar intervals lost elapsed-time range through nanosecond storage and microsecond precision through
Float64seconds. - Design approach: Use a tagged
months/days/microsecondsArrow struct and an exactDecimal(18,6)native kernel. - Correctness / compatibility analysis: The arithmetic matches the inspected Spark implementations across supported versions. Two P2 regressions remain: aggregate nullability normalization breaks interval recognition, and interval hashes now silently use struct semantics.
- Key design decisions: The metadata marker distinguishes intervals from ordinary structs while reusing existing vector infrastructure. Downstream normalization and hashing also need to respect that logical type.
- Implementation sketch: Update Scala/native serde, Arrow writers and codegen, switch
make_intervalto native execution, and extend boundary and round-trip coverage. - Behavioral changes worth calling out: Native execution becomes the default and preserves Spark’s full microsecond range. The reported benchmark improvements concern earlier revisions and were not independently remeasured at this head.
- Suggested improvements: Address both findings and add their aggregate-consumption and hashing regression cases.
Reviewed the entire 23-file diff from 58ab5f618e1e715dee06165424672fddfb90539a9bcf1 to 2074ce9fd26cd867540a2362cddfb90539a9bcf1. The PR remains non-draft. Read the supplied reviews, issue comments and thread snapshot. Routed skills: review-comet-pr, audit-comet-expression, implement-comet-expression, and optimize-comet-expression.
Exact-head CI: 23 successful checks and 14 skipped, including successful required checks, native tests, Spark 4.1 execution suites and TPC validation. Spark SQL compatibility suites and benchmark checks were skipped.
Validation: The focused native microsecond test passed. Disposable native probes and an isolated Spark 4.1.3/Arrow codegen harness reproduced both findings and checked the previous representation. Relevant Spark sources and tests were verified against upstream tags. No full local Maven/JNI integration build or benchmark rerun was performed. No tracked project files or GitHub state were changed.
| val children = field.getChildren | ||
| def child(index: Int, name: String, bits: Int): Boolean = { | ||
| val f = children.get(index) | ||
| f.getName == name && f.getType == new ArrowType.Int(bits, true) && !f.isNullable |
There was a problem hiding this comment.
[P2] Preserve interval recognition after aggregate nullability normalization. The new !f.isNullable requirement rejects schemas produced by coerce_collect_child_nullability: native collect_list makes all three interval children nullable while retaining the marker. With native aggregation and codegen enabled, SELECT transform(collect_list(make_interval(y)), x -> x) FROM t, where t.y contains 1, should return [1 years]. Instead, the aggregate output is classified as array<struct<...>>, so its codegen consumer lacks getInterval and throws. The previous interval representation survives this path. Keep tagged intervals atomic during normalization, or make recognition tolerate this widening, and cover aggregate-to-codegen consumption.
Evidence: Ran the HEAD SparkMakeInterval kernel, the exact planner nullability helper and CometCollectList, then exported the actual aggregate result through Arrow IPC. Its marked children were nullable. HEAD Utils.fromArrowField returned ArrayType(StructType(...),true), and the unchanged HEAD codegen implementation evaluating Spark ArrayTransform threw UnsupportedOperationException: InputArray_col0: getInterval not implemented for this array shape. The equivalent previous IntervalMonthDayNano representation was recognized as ArrayType(CalendarIntervalType,true) and completed successfully. Spark 4.1.3 returned [1 years].
| CALENDAR_INTERVAL_STRUCT_KEY.to_string(), | ||
| "true".to_string(), | ||
| )])); | ||
| DataType::Struct(Fields::from(vec![ |
There was a problem hiding this comment.
[P2] Keep calendar intervals out of the generic struct hashing path. Returning this struct also changes how the existing native hash and xxhash64 kernels process intervals: they hash children in struct order and ignore the logical-type marker. For a Parquet table t(y INT) containing 1, SELECT hash(make_interval(y)), xxhash64(make_interval(y)) FROM t should match Spark. The new representation instead produces two different values without an error. This materially worsens the previous unsupported-type failure into silent incorrect results. Add marker-aware Spark-compatible handling, or route calendar-interval hashes through Spark until that handling exists.
Evidence: With constant folding excluded, Spark 4.1.3 evaluated the equivalent query over range(1,2) as (-351543533, 604378839101286624). A disposable native test passed HEAD SparkMakeInterval(1,0,0,0,0,0,0) directly to HEAD spark_murmur3_hash and spark_xxhash64, obtaining (-912233426, 3333565817687609978). The unchanged hash macro recursively hashes ordinary struct children. Running it on the previous IntervalMonthDayNano(12,0,0) representation instead returned Unsupported data type in hasher: Interval(MonthDayNano), confirming that the silent wrong-result behavior comes from this representation change.
# Conflicts: # docs/source/user-guide/latest/expressions.md # spark/src/test/scala/org/apache/spark/sql/comet/execution/arrow/CometArrowStreamSuite.scala
3 participants
Which issue does this PR close?
Closes #5279.
Closes #5131.
Rationale for this change
Comet represented Spark
CalendarIntervalTypeas ArrowIntervalMonthDayNano. Converting Spark's microseconds to nanoseconds reduced the valid elapsed-time range by 1,000x, while the nativedatafusion-sparkkernel also coercedDecimal(18,6)seconds toFloat64, losing microsecond precision.Spark represents calendar intervals losslessly as separate months, days, and microseconds. Comet needs the same representation across JVM/native boundaries and exact microsecond arithmetic in the native kernel.
What changes are included in this PR?
CalendarIntervalTypeas a Spark-tagged Arrow struct containingmonths: Int32,days: Int32, andmicroseconds: Int64.datafusion-sparkmake_intervalwrapper with an exactDecimal(18,6)microsecond kernel with Spark-compatible NULL and ANSI/TRY overflow behavior.datafusion-sparkdependency.make_intervalboundary and arity cases with source permalinks.make_intervalis fully native.datafusion-sparkinstead: the upstreammake_intervalhard-codesInterval(MonthDayNano)as its return type, which is itself the 1000x range bug; the lossless fix requires the Comet-specific metadata-tagged struct that only Comet's Arrow/FFI layers can map back toCalendarIntervalType, plus Comet's ANSIfail_on_errorsemantics andDecimal(18,6)seconds. TheFloat64precision half is separately fixable upstream.MakeIntervalmoves fromCometCodegenDispatchto fully-native serde, so no dispatch route remains for it; fallback is directly to Spark. The_dispatchfixtures are deleted accordingly.How are these changes tested?
cargo test --manifest-path native/Cargo.toml -p datafusion-comet-spark-expr preserves_spark_microsecond_range_and_overflowcargo check --manifest-path native/Cargo.toml -p datafusion-comet-spark-exprcargo check --manifest-path native/Cargo.toml -p datafusion-cometcargo fmt --manifest-path native/Cargo.toml --all -- --checkmake coreCometArrowStreamSuiteCalendarInterval round-trip testCometCodegenSuiteCalendarInterval codegen testCometSqlFileTestSuite make_interval: 6/6 passed, 0 ignoredCometSqlFileTestSuite calendar_interval: 1/1 passedgit diff --checkBenchmark
CometDatetimeExpressionBenchmark, 1,048,576 rows, Apple M4, JDK 17. The codegen-dispatch result is from parent commit268849c0c; the full-native and Spark results are from this PR at72997e9e8. Both Comet revisions used optimized native builds and the same query and input.Full native is about 10% faster than codegen dispatch and 2.6x faster than Spark by best time.